package com.iflytek.iot.common.log.log2kafka.config;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @author zurun
 * @date 2020/8/17 16:25:07
 */
@Configuration
@ComponentScan("com.iflytek.iot.common.log.log2kafka")
public class BeanConfig {
    @Value("${iot.log2kafka.kafka.serverAddr}")
    private String serverAddr;

    @Bean(destroyMethod = "close")
    @ConditionalOnExpression(value = "${lpc.kafka.producer.enable:true}")
    public KafkaProducer<String, String> kafkaProducer() {

        Serializer<String> serializer = new StringSerializer();
        Map<String, String> properties = new HashMap<>(16);
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverAddr);
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties, serializer, serializer);
        return kafkaProducer;
    }
}
